﻿using Misakai.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using ZooKeeperNet;

namespace Reflexive.IO
{
    public abstract class KafkaBase : IWatcher
    {
        ManualResetEvent _event;

        public KafkaBase()
        {
            _event = new ManualResetEvent(false);
        }

        protected internal KafkaOptions GetOptions(string endpoint, TimeSpan timeout)
        {
            try
            {
                KafkaOptions options = null;
                Monitor.Enter(_event.SafeWaitHandle);
                _event.Reset();
                using (ZooKeeper zk = new ZooKeeper(endpoint, timeout, this))
                {
                    if (_event.WaitOne(new TimeSpan(0, 0, 60)))
                    {
                        var brokers = zk.GetChildren("/brokers/ids", true);
                        var URIs = brokers.Select(id =>
                        {
                            var status = new Org.Apache.Zookeeper.Data.Stat();
                            var bytes = zk.GetData(String.Format("/brokers/ids/{0}", id.ToString()), true, status);
                            var values = System.Text.ASCIIEncoding.ASCII.GetString(bytes);
                            var item = new ServiceStack.Text.JsonSerializer<Dictionary<string, object>>().DeserializeFromString(values);
                            string host = String.Format("http://{0}:{1}", item["host"], item["port"]);
                            return new Uri(host);
                        })
                        .ToArray();
                        options = new KafkaOptions(URIs) { KafkaConnectionFactory = new Reflexive.IO.Kafka.KafkaConnectionFactory(), ResponseTimeoutMs = new TimeSpan(0, 1, 0) };
                    }
                    else
                    {
                        throw new TimeoutException("Zookeeper Connection Timeout");
                    }
                }
                return options;
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.TraceError(e.ToString());
                throw e;
            }
            finally
            {
                Monitor.Exit(_event.SafeWaitHandle);
            }
            
        }

        public virtual void Process(WatchedEvent @event)
        {
            if (@event.State == KeeperState.SyncConnected)
            {
                _event.Set();
            }
            System.Diagnostics.Trace.TraceInformation(@event.ToString());
        }

    }
}
